package com.bw.ads;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.bean.OrderBean;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;

import static org.apache.flink.table.api.Expressions.$;

/*
15> {"baseProvince":{"area_code":"620000","id":"19","iso_3166_2":"CN-GS","iso_code":"CN-62","name":"甘肃","region_id":"7"},
"orderDetail":{"create_time":"2020-06-15 23:34:25","id":"13171","img_url":"http://47.93.148.192:8080/group1/M00/00/01/rBHu8l-rfvmAIpgZAAIvrX6L9fo612.jpg","order_id":"4922","order_price":"5999.00","sku_id":"1","sku_name":"小米10 至尊纪念版 双模5G 骁龙865 120HZ高刷新率 120倍长焦镜头 120W快充 8GB+128GB 陶瓷黑 游戏手机","sku_num":"1","source_id":"12","source_type":"2402","split_total_amount":"5999.00"},
"orderInfo":{"consignee":"司空有坚","consignee_tel":"13576266678","create_time":"2020-06-15 23:34:25","delivery_address":"第14大街第19号楼5单元192门","expire_time":"2020-06-15 23:49:25","id":"4922","img_url":"http://img.gmall.com/398818.jpg","operate_time":"2020-06-15 23:34:26","order_comment":"描述432833","order_status":"1004","out_trade_no":"712257226924328","province_id":"19","total_amount":"14214.00","trade_body":"华为智慧屏V55i-J 55英寸 HEGE-550B 4K全面屏智能电视机 多方视频通话 AI升降摄像头 银钻灰 京品家电等6件商品","user_id":"460"},
"skuInfo":{"category3_id":"61","create_time":"2020-11-11 22:07:35","id":"1","is_sale":"1","price":"5999","sku_default_img":"http://47.93.148.192:8080/group1/M00/00/01/rBHu8l-rfvmAIpgZAAIvrX6L9fo612.jpg","sku_desc":"小米10 至尊纪念版 双模5G 骁龙865 120HZ高刷新率 120倍长焦镜头 120W快充 12GB+256GB 陶瓷黑 游戏手机","sku_name":"小米10 至尊纪念版 双模5G 骁龙865 120HZ高刷新率 120倍长焦镜头 120W快充 8GB+128GB 陶瓷黑 游戏手机","spu_id":"1","tm_id":"1","weight":"1.00"},
"userInfo":{"birthday":"-320","create_time":"2020-06-15 23:34:19","email":"idlv7o09t@googlemail.com","gender":"F","id":"460","login_name":"idlv7o09t","name":"贺馨艺","nick_name":"馨艺","phone_num":"13359459244","user_level":"3"}}

 */

public class FlinkTM5_5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv);


        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop102:9092");
        properties.setProperty("group.id", "group3");
        DataStream<String> stream = fsEnv.addSource(MyKafkaUtil.getKafkaConsumer("dwd_trade_orders_yk1","group113"));

        SingleOutputStreamOperator<OrderBean> mapDS = stream.map(new MapFunction<String, OrderBean>() {
            @Override
            public OrderBean map(String s) throws Exception {
                JSONObject jo = JSON.parseObject(s);
                return new OrderBean(jo.getJSONObject("orderInfo").getString("id"),
                        jo.getJSONObject("userInfo").getString("id"),
                        jo.getJSONObject("orderInfo").getString("order_status"),
                        jo.getJSONObject("orderInfo").getString("create_time")
                        );
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<OrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner((event, timestamp) -> {
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    try {
                        return sdf.parse(event.getCreateTime()).getTime();
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }));
//        mapDS.print();
        fsTableEnv.createTemporaryView("t_cep",mapDS,  $("id"), $("uid"), $("status"), $("createTime"),$("times").rowtime());
       //1）、使用CEP检测用户是否恶意下单，规则为5分钟内连续下单并取消超过10次；（4分） id createTime status
        /*
                SELECT T.*
            FROM t_cep
            MATCH_RECOGNIZE (
              PARTITION BY uid
              ORDER BY times
              MEASURES
                A.uid as auid,A.status as astatus ,A.createTime as atime,
                B.uid as auid,B.status as astatus ,B.createTime as atime
              PATTERN (A B) WITHIN INTERVAL '10' MINUTE
              DEFINE
                A AS status = '1004',
                B AS status = '1004'
            ) AS T
         */
        //fsTableEnv.sqlQuery("select * from t_cep").execute().print();
        fsTableEnv.sqlQuery("SELECT T.*\n" +
                "            FROM t_cep\n" +
                "            MATCH_RECOGNIZE (\n" +
                "              PARTITION BY uid\n" +
                "              ORDER BY times\n" +
                "              MEASURES\n" +
                "                A.uid as auid,A.status as astatus ,A.createTime as atime\n" +
                "              PATTERN (A{2}) WITHIN INTERVAL '5' MINUTE\n" +
                "              DEFINE\n" +
                "                A AS status = '1004'" +
                "            ) AS T").execute().print();

        //fsEnv.execute();

    }
}
